package com.aiobc.business.consumer;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@Component
@RocketMQMessageListener(topic = "xc_topic",consumerGroup="xc_producer",messageModel = MessageModel.BROADCASTING)
public class MQListener implements RocketMQListener<MessageExt> , RocketMQPushConsumerLifecycleListener, MessageListenerConcurrently {

    private static final String topic = "xc_topic";
    private int total;

    @Override
    public void onMessage(MessageExt message) {
        String msgId = message.getMsgId();
        String msg = new String(message.getBody());
        System.out.println("消息全量信息"+message);
        System.out.println("消息id："+msgId+"消息内容："+msg);


    }

    // todo   该方法重写消息监听器的属性,在项目启动时 注册rocketMQ到spring、监听到消息时都会触发该方法
    @SneakyThrows
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        // 设置消费者消息重试次数
        defaultMQPushConsumer.setMaxReconsumeTimes(3);
//        //设置每次消息拉取的时间间隔 单位 毫秒
//        defaultMQPushConsumer.setPullInterval(1000);
//        //最小消费线程池数
//        defaultMQPushConsumer.setConsumeThreadMin(1);
//        //最大消费线程池数
//        defaultMQPushConsumer.setConsumeThreadMax(10);
        //设置消费者单次批量消费的消息数目上限    默认1
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(10);
        //        设置实例名称
        defaultMQPushConsumer.setInstanceName("mqConsumer-business");
//        defaultMQPushConsumer.registerMessageListener((MessageListenerConcurrently) (list, consumeOrderlyContext) -> {
//
//            //todo 签收，即手动ack，rocketMQ默认就是手动ack
//            list.stream().forEach(messageExt -> System.out.println("重写消息监听器==="+new String(messageExt.getBody())));
//
//            // 2、返回成功表示消费成功，不会进行重试
//            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//
//        });
        //todo SQL过滤
        //     生产者Producer过滤:
        //        // 可以同时设置tag和属性，相互不会影响
        //        Message msg = new Message("topic7","zzz",("5消息过滤按照sql：hello rocketmq").getBytes("UTF-8"));
        //        //为消息添加属性
        //        msg.putUserProperty("vip","1");
        //        msg.putUserProperty("age","25");
        //        msg.putUserProperty("username","zhangsan");
        //todo 消费者Consumer过滤:

        //使用消息选择器来过滤对应的属性，语法格式为类SQL语法
        defaultMQPushConsumer.subscribe(topic, MessageSelector.bySql("age >= 18"));
        defaultMQPushConsumer.subscribe(topic, MessageSelector.bySql("username='zhangsan'"));
        // 并集
        defaultMQPushConsumer.subscribe(topic, MessageSelector.bySql("age > 18 or username='zhangsan'"));
        // 交集
        defaultMQPushConsumer.subscribe(topic, MessageSelector.bySql("age > 18 and username='zhangsan'"));
        // 枚举tag
        defaultMQPushConsumer.subscribe(topic, MessageSelector.bySql("TAGS in ('xxx','yyy')"));

        defaultMQPushConsumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                try {

                    for (MessageExt ext:msgs) {
                        System.out.println("消费者收到消息内容==="+ext);
                    }
                    // 2、返回成功表示消费成功，不会进行重试
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

                } catch (Exception e) {
                    System.out.println("result===异常");
                    e.printStackTrace();
                    // 3、返回 RECONSUME_LATER 表示消息需要重试（返回NULL也是一样）
                    // RECONSUME_LATER：通过单词我们知道是 稍后重新消费的意思，即重试
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        setTotal();
        for (MessageExt msg : msgs) {
            System.out.println("========收到消息："+new String(msg.getBody())+"---------总数："+total);
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    public synchronized void setTotal() {
        total++;
    }

    public int getTotal(){
        return total;
    }
}

